/*
 Copyright (C) 2009 QDSS.org
 
 This program is free software: you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
 the Free Software Foundation, either version 3 of the License, or
 any later version.

 This program is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU General Public License for more details.
 
 You should have received a copy of the GNU General Public License
 along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
package org.qdss.persist.engine;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.ObjectUtils;
import org.apache.commons.lang.Validate;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.qdss.persist.DataAccessException;
import org.qdss.persist.Entity;
import org.qdss.persist.Field;
import org.qdss.persist.PersistManager;
import org.qdss.persist.PersistManagerFactory;
import org.qdss.persist.Record;
import org.qdss.persist.dialect.Editor;
import org.qdss.persist.dialect.FieldType;
import org.qdss.persist.engine.entry.ID;
import org.qdss.persist.exception.SQLExceptionConverter;
import org.qdss.persist.metadata.CascadeModel;
import org.qdss.persist.util.SqlHelper;
import org.springframework.jdbc.datasource.DataSourceUtils;

/**
 * 
 * 
 * @author <a href="mailto:zhaofang123@gmail.com">FANGFANG ZHAO</a>
 * @since 0.1.0, Feb 10, 2009
 * @version $Id: PersistManagerImpl.java 20 2009-02-10 03:35:10Z
 *          zhaofang123@gmail.com $
 */
public class PersistManagerImpl extends JdbcSupport implements PersistManager {
	private static final long serialVersionUID = 1204654171280975411L;
	
	private static final Log log = LogFactory.getLog(PersistManagerImpl.class);
	
	transient private PersistManagerFactory managerFactory;
	
	public PersistManagerImpl(PersistManagerFactory managerFactory) {
		super();
		this.managerFactory = managerFactory;
	}

	public PersistManagerFactory getPersistManagerFactory() {
		return managerFactory;
	}

	public Record save(final Record record) throws DataAccessException {
		Validate.isTrue(record.getPrimary() == null);
		
		Entity e = record.getEntity();
		final ID eId = ID.newId(e.getEntityCode());
		
		StringBuilder insert = new StringBuilder("insert into ");
		insert.append(quote(e.getPhysicalName())).append(" ( ").append(quote(e.getPrimaryField().getPhysicalName()));
		StringBuilder valuesclause = new StringBuilder("values ( ?");
		
		final List<Field> fieldList = new LinkedList<Field>();
		fieldList.add(e.getPrimaryField());
		for (Iterator<String> iter = record.getAvailableFieldIterator(); iter.hasNext(); ) {
			Field field = e.getField( iter.next() );
			fieldList.add( field );
			
			insert.append(", ").append(quote(field.getPhysicalName()));
			valuesclause.append(", ?");
		}
		insert.append(" ) ").append(valuesclause.append(" )"));
		
		final String sql = insert.toString();
		valuesclause = null;
		insert = null;
		
		int affected = 0;
		final List<String> refsSqls = new ArrayList<String>();
		try {
			affected = execute(new StatementCallback(){
				public String getSql() {
					return sql;
				}
				
				public Object doInParameters(PreparedStatement pstmt) throws SQLException {
					int index = 1;
					for (Field field : fieldList) {
						Editor editor = field.getType().getFieldEditor();
						Object value = null;
						if (field.getType() == FieldType.PRIMARY) {
							value = eId;
						} else if (field.getType() == FieldType.NREFERENCE) {
							value = eId;
							ID[] array = record.getIDArray(field.getName());
							String sqls[] = ComplexFieldSQLs.formatReferencesMappingUpdateSqls(field, eId, array, false);
							CollectionUtils.addAll(refsSqls, sqls);
						} else {
							value = record.getObjectValue(field.getName());
						}
						editor.set(pstmt, index++, value);
					}
					return null;
				}
			});
			
			if (!refsSqls.isEmpty()) {
				int[] batchExec = executeBatch(refsSqls.toArray(new String[refsSqls.size()]));
				
				for (int be : batchExec)
					affected += be;
			}
		} catch (SQLException sqlex) {
			throw SQLExceptionConverter.convert(sqlex, "#INSERT", sql);
		}
		
		if (log.isDebugEnabled())
			log.debug("total affected " + affected + " rows");
		
		record.setID(e.getPrimaryField().getName(), eId);
		return record;
	}

	public Record update(final Record record) throws DataAccessException {
		Validate.notNull(record.getPrimary());
		
		Entity e = record.getEntity();
		
		StringBuilder update = new StringBuilder("update ");
		update.append(quote(e.getPhysicalName())).append(" set ");
		
		final List<Field> fieldList = new LinkedList<Field>();
		Iterator<String> iter = record.getAvailableFieldIterator();
		boolean first = true;
		while(iter.hasNext()) {
			Field field = e.getField(iter.next());
			if (field.getType() == FieldType.PRIMARY)
				continue;
			
			fieldList.add(field);
			if (field.getType() == FieldType.NREFERENCE)
				continue;
			
			if (first) 
				first = false;
			else 
				update.append(", ");
			update.append(quote(field.getPhysicalName())).append(" = ?");
		}
		update.append(" where ").append(quote(e.getPrimaryField().getName())).append(" = ?");
		fieldList.add(e.getPrimaryField());
		
		final String sql = update.toString();
		update = null;
		
		int affected = 0;
		try {
			if (!first) {
				affected = execute(new StatementCallback(){
					public String getSql() {
						return sql;
					}
					
					public Object doInParameters(PreparedStatement pstmt) throws SQLException {
						int index = 1;
						for (Field field : fieldList) {
							if (field.getType() == FieldType.NREFERENCE)
								continue;
							
							Editor editor = field.getType().getFieldEditor();
							Object value = record.getObjectValue(field.getName());
							if (value == ObjectUtils.NULL)
								pstmt.setNull(index++, managerFactory.getDialect().getSQLType(editor.getType()));
							else
								editor.set(pstmt, index++, value);
						}
						return null;
					}
				});
			}
			
			List<String> refsSqls = new ArrayList<String>();
			for (Field field : fieldList) {
				if (field.getType() != FieldType.NREFERENCE)
					continue;
				
				String[] sqls = ComplexFieldSQLs.formatReferencesMappingUpdateSqls(
						field, record.getPrimary(), record.getIDArray(field.getName()), true);
				CollectionUtils.addAll(refsSqls, sqls);
			}
			
			if (!refsSqls.isEmpty()) {
				int[] batchAffected = executeBatch(refsSqls.toArray(new String[refsSqls.size()]));
				
				for (int ba : batchAffected)
					affected += ba;
			}
		} catch (SQLException sqlex) {
			throw SQLExceptionConverter.convert(sqlex, "#UPDATE", sql);
		}
		if (log.isDebugEnabled())
			log.debug("total affected " + affected + " rows");
		
		return record;
	}

	public Record saveOrUpdate(Record record) throws DataAccessException {
		return (record.getPrimary() == null) ? save(record) : update(record);
	}

	public int delete(final ID id) throws DataAccessException {
		Validate.notNull(id);
		
		Entity entity = managerFactory.getMetadataFactory().getEntity(id.getEntityCode());
		Field[] referenceTos = entity.getReferenceToList();
		
		final String delete = MessageFormat.format("delete from {0} where {1} = ''{2}''",
				quote(entity.getPhysicalName()),
				quote(entity.getPrimaryField().getPhysicalName()),
				id.toLiteral());
		
		final List<String> sqlList = new ArrayList<String>();
		sqlList.add(delete);
		
		Map<Field, ID[]> refToIdMap = null;
		if (referenceTos.length > 0) {
			refToIdMap = new HashMap<Field, ID[]>();
			
			for (Field rto : referenceTos) {
				if (rto.getCascadeModel() == CascadeModel.Ignore)
					continue;
				
				ID[] referenceToIds = getReferenceToIds(rto, id);
				if (referenceToIds.length > 0)
					refToIdMap.put(rto, referenceToIds);
			}
			
			Set<ID> nRefIds = new HashSet<ID>();
			
			for (Iterator<Map.Entry<Field, ID[]>> iter = refToIdMap.entrySet().iterator(); iter.hasNext(); ) {
				Map.Entry<Field, ID[]> e = iter.next();
				Field cField = e.getKey();
				
				// TASK 迭代级联
				if (cField.getType() == FieldType.REFERENCE) {
					String formatted = null;
					if (cField.getCascadeModel() == CascadeModel.Delete) {
						formatted = "delete from {0} where {1} = ''{2}''";
					} else if (cField.getCascadeModel() == CascadeModel.RemoveLinks) {
						// set {1} = null, but some column can not be null
						formatted = "update {0} set {1} = '''' where {1} = ''{2}''";
					} else {
						log.warn("Unknow CascadeModel: " + cField.getCascadeModel());
					}
					
					if (formatted != null) {
						String cql = MessageFormat.format(formatted,
								quote(cField.getOwnEntity().getPhysicalName()),
								quote(cField.getPhysicalName()),
								id.toLiteral() );
						sqlList.add(cql);
					}
				} else {  // Type.NREFERENCE
					CollectionUtils.addAll(nRefIds, e.getValue());
				}
			}
			
			if (!nRefIds.isEmpty()) {
				StringBuilder ql = new StringBuilder("delete from ").append(quote("ReferencesMapping"));
				ql.append(" where ").append(quote("uniqueId")).append(" in ( ");
				
				boolean first = true;
				for (ID rId : nRefIds) {
					if (first)
						first = false;
					else
						ql.append(", ");
					ql.append('\'').append(rId.toLiteral()).append('\'');
				}
				ql.append(" )");
				sqlList.add(ql.toString());
			}
		}
		
		int[] batchAffected = null;
		try {
			batchAffected = executeBatch(sqlList.toArray(new String[sqlList.size()]));
		} catch (SQLException sqlex) {
			throw SQLExceptionConverter.convert(sqlex, "#DELETE", null);
		}
		
		int affected = 0;
		for (int ba : batchAffected)
			affected += ba;
		
		if (log.isDebugEnabled())
			log.debug("total affected " + affected + " rows");
		return affected;
	}

	public int[] delete(ID[] ids) throws DataAccessException {
		int[] rowsAffected = new int[ids.length];
		
		for (int i = 0; i < ids.length; i++) {
			rowsAffected[i] = delete(ids[i]);
		}
		return rowsAffected;
	}
	
	@Override
	protected Connection getConnection() {
		return DataSourceUtils.getConnection(managerFactory.getDataSource());
	}
	
	@Override
	protected void releaseConnection(Connection connect) {
		SqlHelper.release(connect, managerFactory.getDataSource());
	}
	
	
	// ----------------------------------------------------------------------------------------
	
	String quote(String ident) {
		return this.managerFactory.getDialect().quote(ident);
	}
	
	ID[] getReferenceToIds(Field referenceToField, ID masterId) {
		String sql = null;
		Entity own = referenceToField.getOwnEntity();
		if (referenceToField.getType() == FieldType.REFERENCE) {
			sql = MessageFormat.format("select {0} from {1} where {2} = ''{3}'' and {2} is not null",
					quote(own.getPrimaryField().getPhysicalName()),
					quote(own.getPhysicalName()),
					quote(referenceToField.getPhysicalName()),
					masterId.toLiteral());
		} else {  // Type.NREFERENCE
			sql = MessageFormat.format("select {0} from {1} where {2} = {3} and {4} = ''{5}'' and {6} = ''{7}''",
					quote("uniqueId"),
					quote("ReferencesMapping"),
					quote("entity"), own.getEntityCode(),
					quote("field"), referenceToField.getName(),
					quote("referenceId"), masterId.toLiteral());
		}
		
		List<ID> ids = new ArrayList<ID>();
		ResultSet rs = null;
		try {
			rs =  (ResultSet) nativeQuery(sql);
			while (rs.next()) {
				String val = rs.getString(1);
				ids.add( ID.valueOf(val) );
			}
		} catch (SQLException sqlex) {
			throw SQLExceptionConverter.convert(sqlex, "#GET_REFERENCE_TO_IDS", sql);
		} finally {
			SqlHelper.close(rs);
			try {
				if (rs != null)
					SqlHelper.close(rs.getStatement());
			} catch (SQLException e) { /* ignore */ }
		}
		
		if (ids.isEmpty())
			return ID.EMPTY_ID_ARRAY;
		return ids.toArray(new ID[ids.size()]);
	}
}
